;; This file is part of GNUnet
;; Copyright (C) 2004-2013, 2016, 2021, 2022 GNUnet e.V.
;;
;; GNUnet is free software: you can redistribute it and/or modify it
;; under the terms of the GNU Affero General Public License as published
;; by the Free Software Foundation, either version 3 of the License,
;; or (at your option) any later version.
;;
;; GNUnet is distributed in the hope that it will be useful, but
;; WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;; Affero General Public License for more details.
;;
;; You should have received a copy of the GNU Affero General Public License
;; along with this program.  If not, see <http://www.gnu.org/licenses/>.
;;
;; SPDX-License-Identifier: AGPL-3.0-or-later

;; Author: Christian Grothoff
;; Author: Nathan Evans
;; ^^ TODO: not visible yet, but once more parts are ported ...
;; Author: Maxime Devos (Scheme port)
(define-library (gnu gnunet dht client)
  (export %effective-minimum-replication-level
	  %effective-maximum-replication-level
	  %minimum-replication-level
	  %maximum-replication-level
	  bound-replication-level

	  ;; Non-interactive data structures
	  %max-datum-value-length
	  &overly-large-datum make-overly-large-datum overly-large-datum?
	  overly-large-datum-type overly-large-datum-length

	  %overly-large-paths make-overly-large-paths overly-large-paths?
	  overly-large-paths-datum-length
	  overly-large-paths-get-path-length overly-large-paths-put-path-length

	  &malformed-path make-malformed-path malformed-path?
	  malformed-path-what malformed-path-size

	  make-datum datum? datum-type datum-key datum-value datum-expiration
	  datum->insertion insertion? insertion->datum
	  insertion-desired-replication-level
	  make-query query? query-type query-key query-desired-replication-level
	  datum->search-result search-result? search-result->datum
	  search-result-get-path search-result-put-path

	  copy-datum copy-search-result copy-insertion copy-query

	  ;; Network message manipulation procedures
	  ;; (these belong to (gnu gnunet dht network)).
	  (rename (construct-client-get #{ construct-client-get}#)
		  (construct-client-get-stop #{ construct-client-get-stop}#)
		  (construct-client-put #{ construct-client-put}#)
		  (construct-client-result #{ construct-client-result}#)
		  (analyse-client-get #{ analyse-client-get}#)
		  (analyse-client-get-stop #{ analyse-client-get-stop}#)
		  (analyse-client-put #{ analyse-client-put}#)
		  (analyse-client-result #{ analyse-client-result}#))

	  connect
	  disconnect!
	  put!
	  cancel-put!
	  start-get!
	  filter-get:known-results!
	  stop-get!
	  ;; Extended API: monitor
	  start-monitor!
	  stop-monitor!)
  (import (gnu extractor enum)
	  (gnu gnunet block)
	  (gnu gnunet hashcode struct)
	  (gnu gnunet mq)
	  (gnu gnunet mq handler)
	  (gnu gnunet mq-impl stream)
	  (gnu gnunet mq envelope)
	  (only (gnu gnunet server)
		maybe-send-control-message!* make-error-handler)
	  (only (guile)
		pk define-syntax-rule define* lambda* error
		make-hash-table hashq-set! hashq-remove! hashv-set! hashv-ref
		hashv-remove! hash-clear! hash-map->list
		->bool and=>)
	  (only (ice-9 atomic)
		make-atomic-box atomic-box-ref atomic-box-set!)
	  (only (ice-9 match)
		match)
	  (only (ice-9 weak-vector)
		weak-vector weak-vector-ref weak-vector?)
	  (only (gnu extractor enum)
		symbol-value)
	  (only (fibers)
		spawn-fiber)
	  (only (fibers conditions)
		make-condition signal-condition! wait-operation wait)
	  (only (fibers operations)
		perform-operation choice-operation wrap-operation)
	  (only (fibers channels)
		make-channel put-operation get-operation get-message put-message)
	  (only (gnu gnunet mq error-reporting)
		report-error)
	  (only (gnu gnunet concurrency lost-and-found)
		make-lost-and-found collect-lost-and-found-operation)
	  (gnu gnunet dht struct)
	  (only (gnu gnunet message protocols)
		message-type)
	  (only (gnu gnunet netstruct syntactic)
		read% sizeof set%! select)
	  (only (gnu gnunet utils bv-slice)
		slice-length slice/read-only make-slice/read-write slice-copy!
		slice-slice verify-slice-readable slice-copy/read-write
		slice-copy/read-only)
	  (gnu gnunet utils hat-let)
	  (only (gnu gnunet utils cut-syntax)
		cut-syntax)
	  (only (rnrs base)
		and >= = quote * / + - define begin ... let*
		quote case else values apply let cond if > eq?
		<= expt assert exact? integer? lambda for-each
		not expt min max div-and-mod positive? define-syntax
		vector cons)
	  (only (rnrs control)
		unless when)
	  (only (rnrs records syntactic)
		define-record-type)
	  (only (rnrs conditions)
		&error condition make-who-condition define-condition-type)
	  (only (rnrs exceptions)
		raise))
  (begin
    ;; The minimal and maximal replication levels the DHT service allows.
    ;; While the service won't reject replication levels outside this range,
    ;; it will clip them to within this range, so choosing replication levels
    ;; outside this range is useless.
    ;;
    ;; Also, GNUnet v0.15.3 and earlier has a bug where the DHT service can crash
    ;; if the replication level 0 is passed, see https://bugs.gnunet.org/view.php?id=7029.
    ;;
    ;; These values are based on the MINIMUM_REPLICATION_LEVEL and
    ;; MAXIMUM_REPLICATION_LEVEL values in src/dht/gnunet-service-dht_neighbours.c
    ;; of the C implementation.
    (define %effective-minimum-replication-level 1)
    (define %effective-maximum-replication-level 16)
    (define %minimum-replication-level 0)
    (define %maximum-replication-level (- (expt 2 32) 1))

    ;; Called by 'send-get!'.
    (define (bound-replication-level replication-level)
      "Bound the replication level @var{replication-level}, which must be a
valid replication to the level, to the range the DHT service likes."
      (unless (<= %minimum-replication-level replication-level
		  %maximum-replication-level)
	(error "replication level is out of bounds"))
      ;; OOPS swap them
      (max %effective-minimum-replication-level
	   (min %effective-maximum-replication-level replication-level)))

    (define (validate-key key)
      "If @var{key} is, in-fact, a readable /hashcode:512, return it as a
readable bytevector slice. If not, raise an appropriate exception.  The 'what'
in the @code{&missing-capabilities} condition, if any, is the symbol
@code{key}."
      (verify-slice-readable 'key key)
      (if (= (slice-length key) (sizeof /hashcode:512 '()))
	  (slice/read-only key)
	  (error "length of key incorrect")))

    (define (validate-datum datum)
      "If @var{datum} is, in-fact, a datum, return it. Otherwise, raise an
appropriate exception."
      (if (datum? datum) datum (error "not a datum")))

    ;; TODO: more-or-less copied from gnunet_util_lib.h
    (define %max-message-size 65535)

    (define %max-datum-value-length
      (- %max-message-size (sizeof /:msg:dht:client:put '())))
    (assert (<= 0 %max-datum-value-length))

    ;; TODO: maybe check types in 'make-overly-large-datum'.
    ;; TODO: &error / &serious / &condition?
    (define-condition-type &overly-large-datum &error
      make-overly-large-datum
      overly-large-datum?
      ;; block type, as an (exact) integer
      (type   overly-large-datum-type)
      ;; length of the (overly large) value
      (length overly-large-datum-length))

    ;; A condition indicating that the combination of get path, put path
    ;; and datum is too long -- it is the combination that is too long,
    ;; not necessarily any part in particular.
    ;;
    ;; TODO: see &overly-large-datum
    (define-condition-type &overly-large-paths &error
      make-overly-large-paths
      overly-large-paths?
      ;; length of datum value (does not include the key or type)
      (datum-length overly-large-paths-datum-length)
      ;; The length (not the size!) of the get path, i.e. the number
      ;; of path elements.  Zero if there is no get path.
      (get-path-length overly-large-paths-get-path-length)
      ;; Likewise, for the put path.
      (put-path-length overly-large-paths-put-path-length))

    ;; The would-be get-path or put-path does not have the correct size
    ;; to be a path.
    (define-condition-type &malformed-path &error
      make-malformed-path
      malformed-path?
      ;; the symbol 'get-path' or 'put-path'
      (what malformed-path-what)
      ;; size of the would-be path (in octets)
      (size malformed-path-size))

    ;; An key-value entry in the DHT.
    (define-record-type (<datum> make-datum datum?)
      (fields (immutable type datum-type)
	      (immutable key datum-key)
	      (immutable value datum-value)
	      (immutable expiration datum-expiration))
      (protocol
       (lambda (%make)
	 (lambda* (type key value #:key (expiration 0)) ; TODO default expiration
	   "Make a datum object of block type @var{type} (or its corresponding
numeric value), with key @var{key} (a readable @code{/hashcode:512} bytevector
slice), value @var{value} (a readable bytevector slice) and expiring at @var{expiration}.
The keyword argument @var{expiration} is optional, see ???.

The numeric value of the block type can be retrieved with the accessor
@code{datum-type}. The accessors @code{datum-key}, @code{datum-value} and
@code{datum-expiration} return the key, value and expiration time respectively.
It can be tested if an object is a datum object with the predicate @code{datum?}.

The length of @var{value} may be at most @code{%max-datum-value-length}.
If this bound is exceeded, an appropriate @code{&overly-large-datum} and
@code{&who} condition is raised."
	   (define t (canonical-block-type type))
	   (unless (<= (slice-length value) %max-datum-value-length)
	     (raise (condition
		     (make-who-condition 'make-datum)
		     (make-overly-large-datum t (slice-length value)))))
	   (%make t
		  (validate-key key)
		  (slice/read-only value)
		  expiration))))) ; TODO validate expiration

    (define (copy-datum old)
      "Make a copy of the datum @var{old}, such that modifications to the
slices in @var{old} do not impact the new datum."
      (make-datum (datum-type old)
		  (slice-copy/read-only (datum-key old))
		  (slice-copy/read-only (datum-value old))
		  #:expiration (datum-expiration old)))

    ;; A request to insert something in the DHT.
    (define-record-type (<insertion> datum->insertion insertion?)
      (fields (immutable datum insertion->datum)
	      (immutable desired-replication-level
			 insertion-desired-replication-level))
      (protocol
       (lambda (%make)
	 (lambda* (datum #:key (desired-replication-level 3)) ; TODO defaults
	   "Make an insertion object for inserting the datum @var{datum},
desiring a replication level @var{desired-replication-level} (see ??).

The datum and desired replication level can be recovered with the accessors
@var{insertion->datum} and @var{insertion-desired-replication-level}. It can
be tested if an object is an insertion object with the predicate
@code{insertion?}."
	   (%make (validate-datum datum)
		  (bound-replication-level desired-replication-level))))))

    (define (copy-insertion old)
      "Make a copy of the insertion @var{old}, such that modifications to the
slices in @var{old} do not impact the new insertion."
      (datum->insertion (copy-datum (insertion->datum old))
			#:desired-replication-level
			(insertion-desired-replication-level old)))

    (define-record-type (<query> make-query query?)
      (fields (immutable type query-type)
	      (immutable key query-key)
	      (immutable desired-replication-level query-desired-replication-level))
      (protocol
       (lambda (%make)
	 (lambda* (type key #:key (desired-replication-level 3))
	   "Make a query object for searching for a value of block type @var{type}
(or its corresponding numeric value), with key @var{key} (a readable
@code{/hashcode:512} bytevector slice), at desired replication level
@var{desired-replication-level}.

The numeric value of the block type, the key and the desired replication level
can be recovered with the accessors @code{query-type}, @code{query-key} and
@code{query-desired-replication-level}. It can be tested if an object is a
query object with the predicate @code{query?}."
	   (%make (canonical-block-type type)
		  (validate-key key)
		  (bound-replication-level desired-replication-level))))))

    (define (copy-query old)
      "Make a copy of the query object @var{old}, such that modifications to the
slices in @var{old} do not impact the new query object."
      (make-query (query-type old)
		  (slice-copy/read-only (query-key old))
		  #:desired-replication-level
		  (query-desired-replication-level old)))

    (define-record-type (<search-result> datum->search-result search-result?)
      (fields (immutable datum search-result->datum)
	      (immutable get-path search-result-get-path)
	      (immutable put-path search-result-put-path))
      (protocol
       (lambda (%make)
	 (lambda* (datum #:key (get-path #f) (put-path #f))
	   "Make a search result object for the datum @var{datum}. The datum can
be recovered with the accessor @code{search-result->datum}. It can be tested if
an object is a search result with the predicate @code{search-result?}. The
optional arguments @var{get-path} and @var{put-path}, when not false, are bytevector
slices consisting of a list of @code{/dht:path-element}.

The @var{get-path} , if any, is the path from the storage location to the
current peer. Conversely, the @var{put-path}, if any, is a path from the
peer that inserted the datum into the DHT to the storage location. The
@var{get-path}} and @var{put-path} can be accessed with
@code{search-result-get-path} and @code{search-result-put-path} respectively.

When the datum, get path and put path together are too large, a
@code{&overly-large-paths} condition is raised. When the
bytevector slice length of @var{get-path} or @var{put-path} is not a
multiple of the size of a path element, then a @code{&malformed-path}
condition is raised."
	   ;; TODO: can a get-path exist without a put-path?
	   (let^ ((! (make-who)
		     (make-who-condition 'datum->search-result))
		  (! datum (validate-datum datum))
		  (!^ (verify-path what path)
		      "Test if @var{path} looks like a get path, put path or
falsehood. If it is false, return @code{#false}, @code{0} and @code{0}.
Otherwise, if it appears to be a valid path, return @var{path} as a readable
bytevector slice, the size of the path and the length of the path.  If
@var{path} is invalid, raise an appropriate exception."
		     ((? (not path)
			 (values #false 0 0))
		      ;; Verify the slice is readable, and make sure the
		      ;; 'what' field of the &missing-capabilities is
		      ;; precise -- we can rely on slice/read-only to
		      ;; perform capability checking, but then the 'what'
		      ;; field wouldn't be correct.
		      (_ (verify-slice-readable what path))
		      ;; Verify the path actually consists of an integral number
		      ;; of /dht:path-element structures.
		      (! size (slice-length path))
		      (<-- (length remainder)
			   (div-and-mod size (sizeof /dht:path-element '())))
		      (? (positive? remainder)
			 (raise (condition
				 (make-who)
				 (make-malformed-path what size)))))
		     ;; We could place an upper bound on the length of
		     ;; @var{path} here, but that's a bit useless because
		     ;; we will verify the total length (get-path + put-path)
		     ;; later anyway.
		     (values (slice/read-only path) size length))
		  ;; Verify both the get-path and the put-path (if any),
		  ;; remove writability and only keep readability.
		  (<-- (get-path get-path-size get-path-length)
		       (verify-path 'get-path get-path))
		  (<-- (put-path put-path-size put-path-length)
		       (verify-path 'put-path put-path))
		  ;; Make sure the get-path, put-path, datum and
		  ;; /:msg:dht:client:result header will fit in a GNUnet
		  ;; message.  TODO: maybe also consider other messages?
		  (! hypothetical-message-size
		     (+ (sizeof /:msg:dht:client:result '())
			get-path-size put-path-size))
		  (? (> hypothetical-message-size %max-message-size)
		     (raise (condition
			     (make-who)
			     (make-overly-large-paths
			      (slice-length (datum-value datum))
			      get-path-length put-path-length)))))
		 (%make datum get-path put-path))))))

    (define (copy-search-result old)
      "Make a copy of the search result @var{old}, such that modifications to the
slices in @var{old} do not impact the new search result."
      (define get-path (search-result-get-path old))
      (define put-path (search-result-put-path old))
      (datum->search-result (copy-datum (search-result->datum old))
			    #:get-path
			    (and get-path (slice-copy/read-only get-path))
			    #:put-path
			    (and put-path (slice-copy/read-only put-path))))

    

    ;;;
    ;;; Constructing and analysing network messages.
    ;;;
    ;;; These procedures are defined here instead of in (gnu gnunet dht network),
    ;;; but only to prevent cycles.
    ;;;

    (define* (construct-client-get query unique-id #:optional (options 0))
      "Create a new @code{/:msg:dht:client:get} message for the query object
 @var{query}, with @var{unique-id} as ‘unique id’ and @var{options} as options."
      (define s (make-slice/read-write (sizeof /:msg:dht:client:get '())))
      (define-syntax set%!/get (cut-syntax set%! /:msg:dht:client:get <> s <>))
      (set%!/get '(header size) (slice-length s))
      (set%!/get '(header type)
		 (value->index (symbol-value message-type msg:dht:client:get)))
      (set%!/get '(options) options)
      (set%!/get '(desired-replication-level)
		 (query-desired-replication-level query))
      (set%!/get '(type) (query-type query))
      (slice-copy! (query-key query) (select /:msg:dht:client:get '(key) s))
      (set%!/get '(unique-id) unique-id)
      s)

    (define* (construct-client-get-stop key unique-id)
      "Create a new @code{/:msg:dht:client:get:stop} message for cancelling a
get request with @var{unique-id} as unique id and @var{key} as key."
      (define s (make-slice/read-write (sizeof /:msg:dht:client:get:stop '())))
      (define-syntax set%!/stop
	(cut-syntax set%! /:msg:dht:client:get:stop <> s <>))
      (set%!/stop '(header size) (slice-length s))
      (set%!/stop '(header type)
		  (value->index
		   (symbol-value message-type msg:dht:client:get:stop)))
      (set%!/stop '(reserved) 0)
      (set%!/stop '(unique-id) unique-id)
      (slice-copy! key (select /:msg:dht:client:get:stop '(key) s))
      s)

    (define* (construct-client-put insertion #:optional (options 0))
      "Create a new @code{/:msg:dht:client:put} message for the insertion
object insertion with @var{options} as options."
      (define datum (insertion->datum insertion))
      (define size/header (sizeof /:msg:dht:client:put '()))
      (define size (+ size/header (slice-length (datum-value datum))))
      (define message
	(make-slice/read-write
	 (+ size/header (slice-length (datum-value datum)))))
      (define header (slice-slice message 0 size/header))
      (define rest (slice-slice message size/header))
      (define-syntax set%!/put (cut-syntax set%! /:msg:dht:client:put <> header <>))
      (set%!/put '(header type)
		 (value->index (symbol-value message-type msg:dht:client:put)))
      (set%!/put '(header size) size)
      (set%!/put '(type) (datum-type datum))
      (set%!/put '(option) options)
      (set%!/put '(desired-replication-level)
		 (insertion-desired-replication-level insertion))
      (set%!/put '(expiration) (datum-expiration datum))
      ;; Copy key-data pair to insert into the DHT.
      (slice-copy! (datum-key datum) (select /:msg:dht:client:put '(key) header))
      (slice-copy! (datum-value datum) rest)
      message)

    (define (construct-client-result search-result unique-id)
      "Create a new @code{/:msg:dht:client:result} message for the search
result object @var{search-result}, with @var{unique-id} as ‘unique id’"
      (let^ ((! datum (search-result->datum search-result))
	     (! get-path (search-result-get-path search-result))
	     (! put-path (search-result-put-path search-result))
	     (! type (datum-type datum))
	     (! key (datum-key datum))
	     (! value (datum-value datum))
	     (! expiration (datum-expiration datum))
	     (! size/header (sizeof /:msg:dht:client:result '()))
	     (! (path-length path)
		(if path
		    (/ (slice-length path) (sizeof /dht:path-element '()))
		    0))
	     (! get-path-length (path-length get-path))
	     (! put-path-length (path-length put-path))
	     (! size (+ size/header
			(slice-length value)
			get-path-length
			put-path-length))
	     (! message (make-slice/read-write size))
	     (! header (slice-slice message 0 size/header))
	     (! rest (slice-slice message size/header)))
	    (define-syntax set%!/result
	      (cut-syntax set%! /:msg:dht:client:result <> header <>))
	    (set%!/result '(header type)
			  (value->index
			   (symbol-value message-type msg:dht:client:result)))
	    (set%!/result '(header size) size)
	    (set%!/result '(type) type)
	    (set%!/result '(get-path-length) get-path-length)
	    (set%!/result '(put-path-length) put-path-length)
	    (set%!/result '(unique-id) unique-id)
	    (set%!/result '(expiration) expiration)
	    (slice-copy! key (select /:msg:dht:client:result '(key) header))
	    ;; TODO: get-path and put path!
	    (slice-copy! value rest)
	    message))

    (define (analyse-client-get message)
      "Return the query object, the unique id and the options corresponding to
the @code{/:msg:dht:client:result} message @var{message}.  Xqueries are
currently unsupported."
      (let^ ((! type (read% /:msg:dht:client:get '(type) message))
	     (! key (select /:msg:dht:client:get '(key) message))
	     (! desired-replication-level
		(read% /:msg:dht:client:get '(desired-replication-level) message))
	     (! unique-id (read% /:msg:dht:client:get '(unique-id) message))
	     (! options (read% /:msg:dht:client:get '(options) message))
	     (! query
		(make-query type key #:desired-replication-level
			    desired-replication-level)))
	    (values query unique-id options)))

    (define (analyse-client-get-stop message)
      "Return the unique id and the key corresponding to the
@code{/:msg:dht:client:stop} message @var{message}."
      (values (read% /:msg:dht:client:get:stop '(unique-id) message)
	      (select /:msg:dht:client:get:stop '(key) message)))

    (define (analyse-client-put message)
      "Return the insertion object and options corresponding to the
@code{/:msg:dht:client:put} message @var{message}."
      (let^ ((! header (slice-slice message 0 (sizeof /:msg:dht:client:put '())))
	     (! type (read% /:msg:dht:client:put '(type) header))
	     (! key (select /:msg:dht:client:put '(key) header))
	     (! value (slice-slice message (sizeof /:msg:dht:client:put '())))
	     (! desired-replication-level
		(read% /:msg:dht:client:put '(desired-replication-level) header))
	     (! expiration
		(read% /:msg:dht:client:put '(expiration) header))
	     (! options
		(read% /:msg:dht:client:put '(option) header))
	     (! datum (make-datum type key value #:expiration expiration))
	     (! insertion
		(datum->insertion datum #:desired-replication-level
				  desired-replication-level)))
	    (values insertion options)))

    (define (analyse-client-result message)
      "Return search result object and unique id for the
@code{/:msg:dht:client:result} message @var{message}."
      (let^ ((! message (slice/read-only message))
	     (! size/header (sizeof /:msg:dht:client:result '()))
	     (! header (slice-slice message 0 size/header))
	     (! rest (slice/read-only message size/header))
	     (! put-path-length
		(read% /:msg:dht:client:result '(put-path-length) header))
	     (! get-path-length
		(read% /:msg:dht:client:result '(get-path-length) header))
	     (! size/path-element (sizeof /dht:path-element '()))
	     (! put-path
		(slice-slice rest 0 (* size/path-element put-path-length)))
	     (! get-path
		(slice-slice rest (* size/path-element put-path-length)
			     (* size/path-element get-path-length)))
	     (! value
		(slice-slice rest (* (sizeof /dht:path-element '())
				     (+ put-path-length get-path-length))))
	     (! datum
		(make-datum
		 (read% /:msg:dht:client:result '(type) header)
		 (select /:msg:dht:client:result '(key) header)
		 value
		 #:expiration
		 (read% /:msg:dht:client:result '(expiration) header)))
	     (! search-result
		(datum->search-result
		 datum #:get-path get-path #:put-path put-path))
	     (! unique-id (read% /:msg:dht:client:result '(unique-id) header)))
	    (values search-result unique-id)))

    

    ;; New operations are communicated to the main event loop
    ;; via the control channel, using 'maybe-send-control-message!'.
    ;; Operations must be put in id->operation-map before sending them
    ;; to the service!
    (define-record-type (<server> make-server server?)
      (parent <losable>)
      ;; terminal-condition: a disconnect has been requested
      (fields (immutable terminal-condition server-terminal-condition)
	      (immutable control-channel server-control-channel)
	      (immutable lost-and-found server-lost-and-found)
	      ;; Atomic box holding an unsigned 64-bit integer.
	      (immutable next-unique-id/box server-next-unique-id/box))
      (protocol (lambda (%make)
		  (lambda ()
		    (define lost-and-found (make-lost-and-found))
		    ((%make lost-and-found) (make-condition) (make-channel)
		     lost-and-found
		     ;; Any ‘small’ natural number will do.
		     (make-atomic-box 0))))))

    (define (maybe-send-control-message! server . message)
      "Send @var{message} to the control channel of @var{server}, or don't
do anything if @var{server} has been permanently disconnected."
      (apply maybe-send-control-message!* (server-terminal-condition server)
	     (server-control-channel server) message))

    (define-record-type (<get> %make-get get?)
      (parent <losable>)
      (fields (immutable server get:server)
	      (immutable found get:iterator) ; procedure accepting <search-result>
	      (immutable query get:query) ; <query>
	      (immutable unique-id get:unique-id)
	      (immutable options get:options)
	      ;; TODO: test if non-lingering actually works.
	      ;;
	      ;; If #false, 'reconnect' does not keep a strong reference to the
	      ;; search object and 'reconnect' will automatically cancel the
	      ;; search when the search object becomes unreachable.
	      ;;
	      ;; If #true, the search will not be automatically cancelled;
	      ;; 'reconnect' keeps a strong reference.
	      (immutable linger? get:linger?))
      (protocol (lambda (%make)
		  (lambda (server found query unique-id options linger?)
		    ;; When not lingering, add this search object to the lost
		    ;; and found, such that it will eventually be cancelled.
		    ((%make (and (not linger?)
				 (server-lost-and-found server)))
		     server found query unique-id options linger?)))))

    (define-record-type (<put> %make-put put?)
      (fields (immutable server put:server)
	      (immutable inserted put:inserted) ; thunk
	      ;; bytevector slice (/:msg:dht:client:put)
	      (immutable message put:message)))

    (define (send-get! mq get)
      "Send a GET message for @var{get}."
      (send-message! mq (construct-client-get (get:query get)
					      (get:unique-id get)
					      (get:options get))))

    (define (send-stop-get! mq get)
      "Send a message for stopping the get operation @var{get}."
      (send-message!
       mq
       (construct-client-get-stop (query-key (get:query get))
				  (get:unique-id get))))

    (define (fresh-id server)
      "Generate a fresh numeric ID to use for communication with @var{server}."
      ;; Atomically increment the ‘next unique id’, but avoid
      ;; overflow (the GNUnet network structures limit the ‘unique id’
      ;; to being less than (expt 2 64)).
      (%%bind-atomic-boxen
       ((next-unique-id (server-next-unique-id/box server) swap!))
       (let loop ((expected next-unique-id))
	 (define desired (+ 1 expected))
	 ;; TODO(low-priority): handle overflow without errors
	 (when (> desired (- (expt 2 64) 1))
	   (error "you overflowed an 64-bit counter."))
	 (define actual (swap! expected desired))
	 (if (= expected actual)
	     ;; Always returning ‘desired’ instead of ‘expected’ would work
	     ;; too.
	     expected
	     (loop actual)))))

    (define (canonical-block-type type)
      "Return the numeric value of the block type @var{type}
(a @code{block-type?} or in-bounds integer)."
      (cond ((and (integer? type) (exact? type))
	     (unless (and (<= 0 type (- (expt 2 32) 1)))
	       (error "block type out of bounds"))
	     type)
	    (#t
	     (assert (block-type? type))
	     (value->index type))))

    (define* (start-get! server query found #:key (linger? #false))
      "Search for data matching query in the DHT. When a datum is found, call
the unary procedure @var{found} on the search result. It is possible to find
multiple data matching a query. In that case, found is called multiple times.
Searching happens asynchronuously; to stop the search, a fresh search object
for controlling the search is returned.

The procedure found is run from the context of server. As such, if @var{found}
blocks, then all operations on server might block. As such, it is recommended
for found to do as little as possible by itself and instead delegate any work
to a separate fiber.

To avoid expensive copies, the implementation can choose to reuse internal
buffers for the slices passed to @var{found}, which could be overwritten after
the call to @var{found}. As such, it might be necessary to make a copy of the
search result, using @lisp{copy-search-result}.

When the boolean @var{linger?} is false (this is the default), the search is
automatically cancelled when the search object becomes unreachable according
to the GC."
      ;; TODO: options, xquery ...
      (define id (fresh-id server))
      (define options 0) ; TODO: allow setting some options
      (when linger? (assert (eq? linger? #true)))
      (define handle (%make-get server found query id 0 (->bool linger?)))
      (maybe-send-control-message! server 'start-get! handle)
      handle)

    (define (stop-get! search)
      "Cancel the get operation @var{search}.  This is an asynchronuous operation;
it does not have an immediate effect.  This is an idempotent operation; cancelling
a search twice does not have any additional effect."
      (maybe-send-control-message! (get:server search) 'stop-search! search)
      (values))

    (define* (put! server insertion #:key (confirmed values))
      "Perform the insertion @var{insertion}. When the datum has been inserted,
the thunk @var{confirmed} is called. A @emph{put object} is returned which can
be used to stop the insertion.

TODO actually call @var{confirmed}"
      ;; Prepare the message to send.  TODO: options
      (define message (construct-client-put insertion))
      (define handle (%make-put server confirmed message))
      ;; TODO: see start-get!
      (maybe-send-control-message! server 'put! handle)
      handle)

    (define-syntax-rule (well-formed?/path-length slice type (field ...) compare)
      "Verify the TYPE message in @var{slice}, which has @var{field ...} ...
(e.g. one or more of get-path-length or put-path-length) and corresponding
/dht:path-element at the end of the message is well-formed -- i.e., check if the length
of @var{slice} corresponds to the size of @var{type} and the get-path-length and
put-path-length.

@var{compare} must be @code{=} if no additional payload follows, or @code{>=}
if an additional payload may follow.  The message type and the size in the
message header is assumed to be correct."
      ;; Warning: slice is evaluated multiple times!
      (and (>= (slice-length slice) (sizeof type '()))
	   (let* ((header (slice/read-only slice 0 (sizeof type '())))
		  (extra-size (- (slice-length slice) (sizeof type '())))
		  (field (read% type '(field) header))
		  ...)
	     (compare extra-size (* (+ field ...) (sizeof /dht:path-element '()))))))

    ;; TODO reduce duplication with (gnu gnunet nse client) --- maybe introduce
    ;; (gnu gnunet client) as in the C implementation?
    (define (disconnect! server)
      "Asynchronuously disconnect from the DHT service and stop reconnecting,
even if not connected.  This is an idempotent operation.  This is an
asynchronuous request; it won't be fulfilled immediately."
      (maybe-send-control-message! server 'disconnect!))

    (define* (connect config #:key (connected values) (disconnected values)
		      (spawn spawn-fiber))
      "Connect to the DHT service, using the configuration @var{config}.  The
connection is made asynchronuously; the optional thunk @var{connected} is called
when the connection has been made.  The connection can break; the optional thunk
@var{disconnected} is called when it does. If the connection breaks, the client
code automatically tries to reconnect, so @var{connected} can be called after
@var{disconnected}.  This procedure returns a DHT server object."
      (define old-id->operation-map (make-hash-table))
      (define server (make-server))
      ;; We could do @code{(spawn (lambda () (reconnect ...)))} here instead,
      ;; but that causes ‘(DHT) garbage collectable’ to fail.
      (spawn-procedure spawn (server-terminal-condition server) config
		       old-id->operation-map (server-control-channel server)
		       (server-lost-and-found server) #:connected connected
		       #:disconnected disconnected #:spawn spawn)
      server)
    (define (spawn-procedure spawn . rest)
      (spawn (lambda () (apply reconnect rest))))

    ;; TODO: put in new module?
    (define (make-weak-reference to)
      (weak-vector to))
    (define (make-strong-reference to)
      (assert (not (weak-vector? to)))
      to)
    (define (dereference reference)
      (if (weak-vector? reference)
	  (weak-vector-ref reference 0)
	  reference))

    (define* (reconnect terminal-condition config
			old-id->operation-map control-channel lost-and-found
			#:key (spawn spawn-fiber)
			connected disconnected
			#:rest rest)
      ;; The 'id->operation-map' holds get operations that have
      ;; been communicated to the service.  The 'old-id->operation-map'
      ;; is used for reconnecting and holds get operations that need
      ;; to be communicated to the service again.  'old-id->operation-map'
      ;; only shrinks, while 'id->operation-map' can both grow and shrink.
      ;;
      ;; To avoid races, 'id->operation-map' and 'old-id->operation-map'
      ;; are only accessed from 'control'.
      ;;
      ;; To allow cancelling operations when they become unreachable, operations
      ;; are wrapped in a weak reference (unless linger? is #true).  Otherwise,
      ;; they won't ever become unreachable.  Keep in mind that, at least in
      ;; Guile 3.0.7, weak references are broken when the object is returned
      ;; from the guardian (and probably earlier) -- this seems to be a
      ;; difficult to fix bug.
      ;;
      ;; This code is written to support both the correct and incorrect behaviour
      ;; of guardians+weak vectors.
      (define id->operation-map (make-hash-table))
      (define (request-search-result-iterator unique-id)
	"Ask @code{control} what is the iterator for the get operation with
unique id @var{unique-id}.  If there is no such get operation, or the get
operation is cancelled, return @code{#false} instead."
	;; It is possible to look at id->operation-map directly instead,
	;; but hash tables are thread-unsafe.
	;; TODO: reduce allocations
	(define response-channel (make-channel))
	(perform-operation
	 (choice-operation
	  ;; TODO: is this case needed?
	  (wrap-operation (wait-operation terminal-condition)
			  (lambda () #false))
	  (wrap-operation
	   (put-operation control-channel
			  (vector 'request-search-result-iterator unique-id
				  response-channel))
	   (lambda ()
	     (get-message response-channel))))))
      (define handlers
	(message-handlers
	 (message-handler
	  (type (symbol-value message-type msg:dht:monitor:get))
	  ((interpose exp) exp)
	  ((well-formed? slice)
	   ;; The C implementation verifies that 'get-path-length' at most
	   ;; (- (expt 2 16) 1), but this seems only to prevent integer overflow,
	   ;; which cannot happen in Scheme due to the use of bignums.
	   ;;
	   ;; This message does _not_ have a payload, so use = instead of >=.
	   (well-formed?/path-length slice /:msg:dht:monitor:get-response
				     (get-path-length) =))
	  ((handle! slice) ???))
 	 (message-handler
	  (type (symbol-value message-type msg:dht:monitor:get-response))
	  ((interpose exp) exp)
	  ((well-formed? slice)
	   ;; Payload follows, hence >= instead of =.
	   (well-formed?/path-length slice /:msg:dht:monitor:get-response
				     (get-path-length put-path-length) >=))
	  ((handle! slice) ???))
 	 (message-handler
	  (type (symbol-value message-type msg:dht:monitor:put))
	  ((interpose exp) exp)
	  ((well-formed? slice)
	   ;; Payload follows, hence >= instead of =.
	   (well-formed?/path-length slice /:msg:dht:monitor:put
				     (put-path-length) >=))
	  ((handle! slice) ???))
 	 (message-handler
	  (type (symbol-value message-type msg:dht:client:result))
	  ((interpose exp) exp)
	  ((well-formed? slice)
	   ;; Actual data follows, hence >= instead of =.
	   (well-formed?/path-length slice /:msg:dht:client:result
				     (get-path-length put-path-length) >=))
	  ((handle! slice)
	   ;; The DHT service found some data we were looking for.
	   (let^ ((<-- (search-result unique-id)
		       ;; TODO: maybe verify the type and key?
		       (analyse-client-result slice))
		  (! handle (request-search-result-iterator unique-id))
		  (? (not handle)
		     ;; Perhaps the search object became unreachable;
		     ;; 'process-stop-search' (see next commit) will be
		     ;; called soon to inform the DHT service.
		     (values))
		  (? (get? handle)
		     ;; TODO might not be true once monitoring operations
		     ;; are supported.
		     ((get:iterator handle) search-result)))
		 ;; TODO: wrong type (maybe a put handle?).
		 TODO-error-reporting/2)))))
      (define error-handler
	(make-error-handler connected disconnected terminal-condition
			    control-channel))
      (define mq (connect/fibers config "dht" handlers error-handler))
      (define (process-stop-search get)
	;; TODO: tests!
	;; TODO: cancel outstanding messages to the DHT services for this
	;; get operation (including the request to start searching), if
	;; any.
	(hashv-remove! old-id->operation-map (get:unique-id get))
	(when (hashv-ref id->operation-map (get:unique-id get))
	  (hashv-remove! id->operation-map (get:unique-id get))
	  (send-stop-get! mq get)))
      (define loop-operation
	(choice-operation
	 (get-operation control-channel)
	 (wrap-operation (collect-lost-and-found-operation lost-and-found)
			 (lambda (lost) (cons 'lost lost)))))
      (define (control)
	"The main event loop."
	(control* (perform-operation loop-operation)))
      (define (control* message)
	(match message
	  (('oops! key . arguments)
	   ;; Some unknown error, report it (report-error) and close
	   ;; the queue (close-queue!).  'connected' will be called
	   ;; from the 'input:regular-end-of-file' case in 'error-handler'.
	   ;;
	   ;; The error reporting and closing happen in no particular order.
	   (signal-condition! terminal-condition)
	   (apply report-error key arguments)
	   (close-queue! mq)
	   (values))
	  (('disconnect!)
	   ;; Ignore future requests instead of blocking.
	   (signal-condition! terminal-condition)
	   ;; Close networking ports.
	   (close-queue! mq)
	   ;; And the fibers of the <server> object are now done!
	   (values))
	  (('reconnect!)
	   ;; Restart the loop with a new message queue.
	   (apply reconnect terminal-condition config id->operation-map
		  control-channel lost-and-found rest))
	  (('start-get! get)
	   ;; Register the new get operation, such that we remember
	   ;; where to send responses to.
	   (hashv-set! id->operation-map (get:unique-id get)
		       ((if (get:linger? get)
			   make-strong-reference
			   make-weak-reference) get))
	   ;; (Asynchronuously) send the GET message.
	   (send-get! mq get)
	   ;; Continue!
	   (control))
	  (('stop-search! get)
	   (process-stop-search get)
	   ;; Continue!
	   (control))
	  (('put! put)
	   ;; Send the put operation to the DHT service.
	   (send-message! mq (put:message put))
	   ;; Continue!
	   (control))
	  ;; Send by @code{request-search-result-iterator}.
	  (#('request-search-result-iterator unique-id response-channel)
	   (put-message response-channel
			(and=> (hashv-ref id->operation-map unique-id)
			       dereference))
	   ;; Continue!
	   (control))
	  (('resend-old-operations!)
	   ;; Restart old operations.  Only get operations need to be submitted
	   ;; again.
	   ;;
	   ;; TODO: restarting monitoring operations
	   (for-each (lambda (reference)
		       (define get (dereference reference))
		       ;; If the (weak) reference is broken, that means the
		       ;; operation is unreachable, so then there is no point
		       ;; to resending the get operation.
		       (when get
			 (hashv-set! id->operation-map (get:unique-id get)
				     reference)
			 (send-get! mq get)))
		     ;; XXX: @code{hash-for-each} forms a continuation barrier,
		     ;; so turn the hash table into a list before iterating.
		     (hash-map->list (lambda (x reference) reference)
				     old-id->operation-map))
	   ;; Free some memory.
	   (hash-clear! old-id->operation-map)
	   ;; Continue!
	   (control))
	  ;; Some handles became unreachable and can be cancelled.
	  (('lost . lost)
	   (let loop ((lost lost))
	     (match lost
	       ;; Continue!
	       (() (control))
	       ((object . rest)
		(match object
		  ((? get? get)
		   (process-stop-search get)
		   (loop rest))
		  ((? server? server)
		   (control* '(disconnect!))))))))))
      ;; Start the main event loop.
      (control))))
